package com.example.mq.mqserver;

import com.example.mq.common.Consumer;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.*;
import com.example.mq.mqserver.datacenter.DiskDataCenter;
import com.example.mq.mqserver.datacenter.MemoryDataCenter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 通过这个类 来表示虚拟主机
 * 每个虚拟主机下面都管理这自己的 交换机 队列 消息数据 绑定
 * 同时提供API供上层调用
 * 针对 VirtualHost 这个类 作为业务逻辑的整合者 就需要对代码中抛出的异常进行处理
 */
public class VirtualHost {

    private String virtualHostName;
    private MemoryDataCenter memoryDataCenter=new MemoryDataCenter();
    private DiskDataCenter diskDataCenter=new DiskDataCenter();
    private ConsumerManger consumerManger=new ConsumerManger(this);
    private Router router=new Router();

    //操纵交换机的锁对象
    private final Object exchangeLocker=new Object();

    //操作队列的锁对象
    private final Object queueLocker=new Object();

    public String getVirtualHostName() {
        return virtualHostName;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    public VirtualHost(String name)  {
        //初始化
        this.virtualHostName=name;
        //对于内存数据来说 不需要额外进行初始化操作的 只要new对象 就可以了

        //针对数据库文件来说 需要建库建表等操作 进行初始化
        diskDataCenter.init();
        //还需要针对硬盘的数据 恢复到内存中这样的操作
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | MqException | ClassNotFoundException e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 恢复内存数据失败!");
        }
    }

    //创建交换机
    //如果交换机不存在 就创建 如果不存在 就直接返false 成功返回true
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType,
                                   boolean durable, boolean autoDelete, Map<String,Object> arguments){

        //把交换机名字+虚拟主机名字作为前缀
        exchangeName=virtualHostName+exchangeName;
        //1.判断该交换机是否已经存在
        try {
            synchronized (exchangeLocker) {  //防止出现线程安全
                //通过内存查询
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if (existsExchange != null) {
                    //交换机已经存在 直接返回
                    System.out.println("[VirtualHost] 交换机已经存在! exchangeName=" + exchangeName);
                    return true;
                }
                //2.创建交换机
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
                //3.把交换机对象写入硬盘
                if (durable) {
                    //需要持久化处理 才需要存储到硬盘
                    diskDataCenter.insertExchange(exchange);
                }
                //4.把交换机对象写入内存
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 交换机创建完成! exchangeName=" + exchangeName);
                //上述逻辑 先写硬盘 后写内存 因为硬盘更容易写失败 如果硬盘失败了 内存就不写了
                //要是先写内存 如果内存成功了 硬盘失败了 就还需要将内存删除 就比较麻烦了
                }
            }catch(Exception e){
                //处理异常
                System.out.println("[VirtualHost] 交换机创建失败! exchangeName=" + exchangeName);
                e.printStackTrace();
                return false;
            }
            return true;
    }

    //删除交换机
    public boolean exchangeDelete(String exchangeName){
        exchangeName=virtualHostName+exchangeName;
        try{
            synchronized (exchangeLocker) {
                //删除之前 先判断是否存在这个交换机
                Exchange toDelete = memoryDataCenter.getExchange(exchangeName);
                if (toDelete == null) {
                    throw new MqException("[VirtualHost] 交换机不存在!");
                }
                //是持久化存储 就删除硬盘上的数据
                if (toDelete.isDurable()) {
                    diskDataCenter.deleteExchange(exchangeName);
                }
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost] 交换机删除成功! exchangeName =" + exchangeName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机删除失败! exchangeName="+exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    //创建队列
    public boolean queueDeclare(String queueName,boolean durable,boolean exclusive,boolean autoDelete,Map<String,Object> arguments){
        //转换名字
        queueName=virtualHostName+queueName;
        //看队列是否存在
        try{
            synchronized (queueLocker) {
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if (existsQueue != null) {
                    System.out.println("[VirtualHost] 队列已经存在! queueName=" + queueName);
                    return true;
                }
                //如果不存在 就创建队列
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setExclusive(exclusive);
                queue.setDurable(durable);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                //创建硬盘上的队列
                if (durable) {
                    diskDataCenter.insertQueue(queue);
                }
                //创建内存中的队列
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建成功! queueName =" + queueName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列创建失败! queueName =" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    //删除队列
    public boolean queueDelete(String queueName){
        //名字转换
        queueName=virtualHostName+queueName;
        try {
            synchronized (queueLocker) {   //多个线程 调用删除队列 添加队列 就会出现问题
                //1.根据队列名字 查询队列是否存在
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new MqException("[VirtualHost] 队列不存在!");
                }
                //删除硬盘数据
                if (queue.isDurable()) {
                    diskDataCenter.deleteQueue(queueName);
                }
                //删除内存数据
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 队列删除成功! queueName=" + queueName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列删除失败! queueName="+queueName);
            e.printStackTrace();
            return false;
        }
    }

    //创建绑定
    public boolean queueBind(String exchangeName,String queueName,String bindingKey){
        //转换名字
        exchangeName=virtualHostName+exchangeName;
        queueName=virtualHostName+queueName;
        try {
            synchronized (exchangeLocker){   //两个对象都会操作 所以加两把锁
                synchronized (queueLocker){
                    //1.判断此时的绑定是否存在
                    Binding existsBinding =memoryDataCenter.getBinding(exchangeName,queueName);
                    if(existsBinding!=null){
                        //绑定已经存在
                        throw  new MqException("[VirtualHost] 绑定已经存在! queueName = "+queueName+", exchangeName="+exchangeName);
                    }
                    //2.验证 bandingKey 是否合法
                    if(!router.checkBindingKey(bindingKey)){
                        throw  new MqException("[VirtualHost] bindingKey 非法! bindingKey="+bindingKey);
                    }
                    //4.创建一个binding对象
                    Binding binding=new Binding();
                    binding.setBindingKey(bindingKey);
                    binding.setQueueName(queueName);
                    binding.setExchangeName(exchangeName);
                    //5.获取对应的交换机和队列 如果不存在 就无法创建
                    MSGQueue queue=memoryDataCenter.getQueue(queueName);
                    if(queue==null){
                        throw  new MqException("[VirtualHost] 队列不存在! queueName="+ queueName);
                    }
                    Exchange exchange=memoryDataCenter.getExchange(exchangeName);
                    if(exchange==null){
                        throw  new MqException("[VirtualHost] 交换机不存在! exchangeName="+ exchangeName);
                    }
                    //6.创建绑定消息
                    //写入硬盘
                    if(queue.isDurable() && exchange.isDurable()){
                        diskDataCenter.insertBinding(binding);
                    }
                    //写入内存
                    memoryDataCenter.insertBinding(binding);
                    System.out.println("[VirtualHost] 创建绑定成功! queueName = "+queueName+", exchangeName="+exchangeName);
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 绑定创建失败! queueName = "+queueName+", exchangeName="+exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    //删除绑定
    public boolean queueUnbind(String queueName,String exchangeName){
        queueName=virtualHostName+queueName;
        exchangeName=virtualHostName+exchangeName;
        try {
            synchronized (exchangeLocker){
                synchronized (queueLocker){
                    //1.获取 binding 看是否存在
                    Binding binding=memoryDataCenter.getBinding(exchangeName,queueName);
                    if(binding==null){
                        throw  new MqException("[VirtualHost] 删除绑定失败! 绑定不存在! queueName = "+queueName+", exchangeName="+exchangeName);
                    }
                    //2.获取对应的队列和交换机 看是否存在(如果先删除队列/交换机 绑定就删除不掉了所以我们此时就不校验其是否存在了)
//            MSGQueue queue=memoryDataCenter.getQueue(queueName);
//            if(queue==null){
//                throw  new MqException("[VirtualHost] 队列不存在! queueName="+ queueName);
//            }
//            Exchange exchange=memoryDataCenter.getExchange(exchangeName);
//            if(exchange==null){
//                throw  new MqException("[VirtualHost] 交换机不存在! exchangeName="+ exchangeName);
//            }
//            //3.删除绑定关系
//            if(queue.isDurable()&&exchange.isDurable()){
//                diskDataCenter.deleteBinding(binding);
//            }
                    //无论其是否持久化了 都尝试删一下 就算不存在 也没什么副作用
                    diskDataCenter.deleteBinding(binding);
                    //删除内存中的绑定关系
                    memoryDataCenter.deleteBinding(binding);
                    System.out.println("[VirtualHost] 删除绑定成功!");
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 删除绑定失败!");
            e.printStackTrace();
            return false;
        }
    }

    //发送消息到指定的交换机/队列中
    public boolean basicPublish(String exchangeName,String routingKey,BasicProperties basicProperties,byte[] body){
        try {
            //1.转换名字
            exchangeName=virtualHostName+exchangeName;
            //2.检查 routingKey 是否合法
            if(!router.checkRoutingKey(routingKey)){
                throw new MqException("[VirtualHost] routingKey 非法 routingKey="+routingKey);
            }
            //3.找到交换机对象
            Exchange exchange=memoryDataCenter.getExchange(exchangeName);
            if(exchange==null){
                //交换机不存在
                throw new MqException("[VirtualHost] 交换机不存在 exchangeName="+exchangeName);
            }
            //4.判断交换机类型,不同类型发送方式不同
            if(exchange.getType()==ExchangeType.DIRECT){
                //按照直接交换机的方式转发消息
                //以 routingKey 作为队列的名字 直接把消息写入指定的队列中
                //此时 可以无视绑定关系了
                String queueName=virtualHostName+routingKey;
                //5.构造消息对象
                Message message=Message.createMessageWitId(routingKey,basicProperties,body);
                //6.查找该队列名对应的对象
                MSGQueue queue=memoryDataCenter.getQueue(queueName);
                if(queue==null){
                    throw new MqException("[VirtualHost] 队列不存在 queueName="+queueName);
                }
                //7.队列存在 直接写入消息
                sendMessage(queue,message);
            }else {
                //按照 fanout 和 topic 的方式来转发
                //5.找到该交换机关联的所有绑定
                ConcurrentHashMap<String, Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
                for(Map.Entry<String,Binding> entry : bindingsMap.entrySet()){
                    //1.获取到绑定对象 判定对应的队列是否存在
                    Binding binding=entry.getValue();
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if(queue==null){
                        //找不到绑定对应的队列
                        System.out.println("[VirtualHost] basicPublish 发送消息时,发现队列不存在 queueName="+binding.getQueueName());
                        continue;
                    }
                    //2.构造消息对象
                    Message message=Message.createMessageWitId(routingKey,basicProperties,body);
                    //3.判定这个消息是否能转发给队列
                    //如果是 fanout,所有绑定的队列都要转发
                    //如果是 topic 还需要判定一下 bindingKey 和 routingKey 是否一致
                    if(!router.route(exchange.getType(),binding,message)){
                        //不能转发就跳过
                        continue;
                    }
                    //4.真正转发消息给队列
                    sendMessage(queue,message);
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 消息发送失败!");
            e.printStackTrace();
            return false;
        }
    }

    //把消息写入到队列中
    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        //此时的queue在外面已经校验了queue 一定的存在的(针对我们上层调用的方法)
         //写到内存和硬盘上
        int deliverMode = message.getDeliverMode();
        //deliverMode 为1 不持久化 为2 为持久化
        if(deliverMode==2){
            diskDataCenter.sendMessage(queue,message);
        }
        //写入内存
        memoryDataCenter.sendMessage(queue,message);
        //此处还需要 通知消费者可以消费通知了
        consumerManger.notifyConsume(queue.getName());
    }

    //订阅消息(添加一个队列的订阅者 当队列生产消息的时候 就把消息推送给订阅者)
    //consumerTag：消费者的身份标识
    // autoAck 消息被消费完成后 应答的方式 为true 表示自动应答 为 false
    // Consumer 是一个回调方法 此处设定成函数式接口 这样后续调用 basicConsume 并且传实参的时候 就可以写作lambda的样子
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer){
        //构造一个 ConsumerEnv 对象  把这个对应的队列找到 再把这个对象添加到队列中
        queueName=virtualHostName+queueName;
        try {
            consumerManger.addConsumer(consumerTag,queueName,autoAck,consumer);
            System.out.println("[VirtualHost] basicConsume 成功! queueName="+queueName);
            return true;
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("[VirtualHost] basicConsume 失败! queueName="+queueName);
            return false;
        }
    }

    public boolean basicAck(String queueName,String messageId){
        queueName=virtualHostName+queueName;
        try {
            //获取到消息和队列
            Message message=memoryDataCenter.getMessage(messageId);
            MSGQueue queue=memoryDataCenter.getQueue(queueName);
            if(message==null){
                throw new MqException("[VirtualHost] 要确认的消息不存在 messageId="+messageId);
            }
            if(queue==null){
                throw new MqException("[VirtualHost] 要确认的队列不存在 queueName="+queueName);
            }
            //删除硬盘数据
            if(message.getDeliverMode()==2){
                diskDataCenter.deleteMessage(queue,message);
            }
            //删除消息中心数据
            memoryDataCenter.removeMessage(messageId);
            //删除待确认队列数据
            memoryDataCenter.removeMessageWaitAck(queueName,messageId);
            System.out.println("[VirtualHost] basicAck 成功! 消息被成功确认 queueName="+queueName+", messageId"+messageId);
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] basicAck 失败! 消息确认失败 queueName="+queueName+", messageId"+messageId);
            e.printStackTrace();
            return false;
        }
    }

}
